package com.cluster.hadoop.api;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.net.URI;
import java.net.URISyntaxException;

public class HDFSOperation {

    public static final String HSFS_URI = "hdfs://192.168.10.179:9000";
    private static Configuration conf = new Configuration();

    static {
        conf.set("fs.defaultFS", HSFS_URI);
    }


    //hdfs dfs -put file /
    public void uploadFile2HDFS() throws Exception {

        FileSystem fs = FileSystem.get(new URI(HSFS_URI), conf, "root");
        fs.copyFromLocalFile(new Path("C:\\Users\\XBZX\\Desktop\\西安政务诚信监测评价指标V4(2).xlsx"), new Path(HSFS_URI + "/dxg/upload_file/西安政务诚信监测评价指标V4(2).xlsx"));
        fs.close();
    }

    // 从hdfs下载文件到windows系统时，需要拷贝winutils.exe和hadoop.dll文件到C:\windows\system32目录中，否则报错 (null) entry in command string: null chmod 0644
    //hdfs dfs -get /
    public void downloadFileFormHDFS() throws URISyntaxException, IOException, InterruptedException {
//        Configuration conf = new Configuration();
//        conf.set("fs.defaultFS", HSFS_URI);
        FileSystem fs = FileSystem.get(new URI(HSFS_URI), conf, "root");
        fs.copyToLocalFile(new Path(HSFS_URI + "/dxg/西安政务诚信监测评价指标V4(2).xlsx"), new Path("D:/"));
        fs.close();
    }

    //hdfs dfs -rm -f -R /文件
    public void deleteFromHDFS() throws Exception {
//        Configuration conf = new Configuration();
//        conf.set("fs.defaultFS", HSFS_URI);
        FileSystem fs = FileSystem.get(new URI(HSFS_URI), conf, "root");
        fs.delete(new Path("/dxg/upload_file"), true);
        fs.close();
    }

    public void mkdir() throws Exception {
        FileSystem fs = FileSystem.get(new URI(HSFS_URI), conf, "root");
        fs.mkdirs(new Path("/dxg/upload_file"));
        fs.close();
    }

    //hdfs dfs -ls /
    public void listHDFSFile() throws Exception {
        FileSystem fs = FileSystem.get(new URI(HSFS_URI), conf, "root");
        FileStatus[] fileStatuses = fs.listStatus(new Path("/dxg"));
        for (FileStatus fileStatus : fileStatuses) {
            System.out.println(fileStatus.getPath().getName());
        }
        fs.close();

    }

    public void readFile() throws Exception {
        FileSystem fs = FileSystem.get(new URI(HSFS_URI), conf, "root");
        FSDataInputStream open = fs.open(new Path("/dxg/data_news/part-m-00000"));
        //调用方法处理文件
        String s1 = inputToString(open,"UTF-8");
        System.out.println("s1="+s1);
        fs.close();

        String[] split = s1.split("\t");
        for (int i = 0; i < split.length; i++) {
            System.out.println(i + " : " + split[i]);
        }
    }

    //私有的文件处理方法
    private static String inputToString(InputStream inputStream, String encode) {
        try {
            if (encode.equals("") && encode == null) {
                encode = "utf-8";
            }
            //字符串缓冲流来处理字符流中的内容并指定编码
            BufferedReader br = new BufferedReader(new InputStreamReader(inputStream, encode));
            StringBuilder sb = new StringBuilder();
            String ss = "";
            // 循环读取缓冲流中的内容并添加到一个StringBuilder中，读取完一个换行
            while ((ss = br.readLine()) != null) {
                System.out.println(ss);
                sb.append(ss).append("\n");
                break;

            }
            // 以string类型返回读取完毕的内容
            return sb.toString();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    public static void main(String[] args) throws Exception {
        HDFSOperation hdfsOperation = new HDFSOperation();
//        hdfsOperation.uploadFile2HDFS();
//        hdfsOperation.deleteFromHDFS();
//        hdfsOperation.mkdir();
//        hdfsOperation.downloadFileFormHDFS();
//        hdfsOperation.listHDFSFile();
        hdfsOperation.readFile();

    }

}
